package com.education.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.app.BaseAppV1;
import com.education.realtime.bean.UserReview;
import com.education.realtime.common.Constant;
import com.education.realtime.util.AtguiguUtil;
import com.education.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/*
    每个课程的用户 好评数 和参与评价的用户数，总的评价分数
        求出用户总数 和 用户总分数
        一个课程用户要么参与评价 要么没有参与评价
 */
public class DwsInteractionCourseReviewWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsInteractionCourseReviewWindow().init(2015,"Dws_13_InteractionUserCourseReviewWindow","dwd_interaction_review");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 过滤出没有评价的用户
        stream.map(JSON::parseObject)
                .filter(obj->obj.getString("review_stars")!=null)
                .map(new MapFunction<JSONObject, UserReview>() {
                    @Override
                    public UserReview map(JSONObject obj) throws Exception {
                        String courseId = obj.getString("course_id");
                        Long reviewStars = obj.getLong("review_stars");
                        if (reviewStars >= 5){
                            return UserReview.builder().courseId(courseId).userGoodReview(1L).userStars(reviewStars).userReview(1L).ts(obj.getLong("ts")*1000).build();
                        }else{
                            return UserReview.builder().courseId(courseId).userGoodReview(0L).userStars(reviewStars).userReview(1L).ts(obj.getLong("ts")*1000).build();
                        }
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserReview>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((obj, e)->obj.getTs()))
                .keyBy(obj->obj.getCourseId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<UserReview>() {
                    @Override
                    public UserReview reduce(UserReview v1, UserReview v2) throws Exception {
                        v1.setUserGoodReview(v1.getUserGoodReview() + v2.getUserGoodReview());
                        v1.setUserReview(v1.getUserReview() + v2.getUserReview());
                        v1.setUserStars(v1.getUserStars() + v2.getUserStars());
                        return v1;
                    }
                }, new ProcessWindowFunction<UserReview, UserReview, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context ctx, Iterable<UserReview> iterable, Collector<UserReview> out) throws Exception {
                        UserReview next = iterable.iterator().next();
                        next.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                        next.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                        next.setTs(ctx.currentProcessingTime());
                        out.collect(next);
                    }
                })
                .addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB,"dws_interaction_course_review_window",UserReview.class));
    }
}
